package com.aididi.consumer;

import com.aididi.dto.OrderDto;
import com.aididi.enums.ErrorCode;
import com.aididi.exception.BizException;
import com.aididi.utils.DateUtil;
import com.alibaba.fastjson.JSON;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

/**
 * @author ：xuzexiang
 * @description：订单消息消费类
 * @date ： 2021/10/28 10:17
 */
@Slf4j
@Getter
@Setter
public class OrderMessageListener implements MessageListenerConcurrently {
    private String topic;
    private String tag;
    private int reconsume;

    public OrderMessageListener(String topic ,String tag,int reconsume){
        this.topic = topic;
        this.tag  = tag;
        this.reconsume = reconsume;
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if(CollectionUtils.isEmpty(list)){
            log.info("接收消息为空，不处理，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        for (MessageExt messageExt : list) {
            log.info("topic: [{}], tag: [{}] 接收到的消息：{} - {}",this.topic,this.tag,messageExt.getMsgId(),messageExt.getKeys());
            String msg =null ;
            String errorMsg = "";
            if(StringUtils.equals(messageExt.getTopic(),this.topic) && StringUtils.equals(messageExt.getTags(),this.tag)){
                try {
                    msg = new String(messageExt.getBody(),"UTF-8");
                    log.info("mq接收到的消息{}-时间{}，msgId: {}",msg, DateUtil.stampToDateStr(messageExt.getBornTimestamp(),DateUtil.DATE_TIME_PATTERN) ,messageExt.getMsgId());
                    //处理消息
                    OrderDto orderDto = JSON.parseObject(msg,OrderDto.class);

                    continue;
                } catch (BizException e) {
                    log.info("消息消费异常：{}",e.getErrMsg());
                    errorMsg = e.getErrMsg();
                    if(ErrorCode.MQ_NORSEND.getRespCode().equals(e.getErrCode())){
                        messageExt.setReconsumeTimes(100);
                    }
                }catch (Exception e){
                    log.info("消息消费异常：{}",e.getMessage());
                    errorMsg = e.getMessage();
                }
                //RocketMQ默认允许每条消息最多重试16次
                if(messageExt.getReconsumeTimes()+1>this.reconsume){
                    log.info("接收到的消息：[{}]，重试超过{}次数",msg,this.reconsume);
                    //超过重试次数的消息放到消息日志表里

                    continue;
                }
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }
        }
        // ack，只有等上面一系列逻辑都处理完后，到这步CONSUME_SUCCESS才会通知broker说消息消费完成，如果上面发生异常没有走到这步ack，则消息还是未消费状态。而不是像比如redis的blpop，弹出一个数据后数据就从redis里消失了，并没有等我们业务逻辑执行完才弹出。
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
